package api

import bean.SensorReading
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, SplitStream, StreamExecutionEnvironment}

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/14 20:22
  *
  * 多流转换操作
  */
object Transform2 {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputPath: String = "src/main/resources/sensor.txt"
    val inputStream: DataStream[String] = env.readTextFile(inputPath)

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = inputStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })


    //温度大于30度
    val splitStream: SplitStream[SensorReading] = dataStream
      .split(data => {
        if (data.temperature > 30.0)
          Seq("high")
        else
          Seq("low")
      })
    val highTempStream: DataStream[SensorReading] = splitStream.select("high")

    val lowTempStream: DataStream[SensorReading] = splitStream.select("low")

    val allTempStream: DataStream[SensorReading] = splitStream.select("high", "low")



    highTempStream.print("high")
    lowTempStream.print("low")
    allTempStream.print("all")
    env.execute("temperature test")

  }

}
